package sbc.group1.lab1.mozartspaces;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.ConcurrentSkipListMap;

import org.mozartspaces.capi3.Capi3Exception;
import org.mozartspaces.capi3.CoordinationData;
import org.mozartspaces.capi3.CoordinatorLockedException;
import org.mozartspaces.capi3.CountNotMetException;
import org.mozartspaces.capi3.EntryLockedException;
import org.mozartspaces.capi3.IsolationLevel;
import org.mozartspaces.capi3.OperationType;
import org.mozartspaces.capi3.javanative.LockedExceptionsHelper;
import org.mozartspaces.capi3.javanative.coordination.AbstractDefaultSelector;
import org.mozartspaces.capi3.javanative.coordination.ImplicitNativeCoordinator;
import org.mozartspaces.capi3.javanative.coordination.NativeCoordinator;
import org.mozartspaces.capi3.javanative.coordination.NativeCoordinatorTranslator;
import org.mozartspaces.capi3.javanative.coordination.NativeSelectorTranslator;
import org.mozartspaces.capi3.javanative.isolation.Availability;
import org.mozartspaces.capi3.javanative.isolation.NativeIsolationManager;
import org.mozartspaces.capi3.javanative.isolation.NativeSubTransaction;
import org.mozartspaces.capi3.javanative.operation.NativeEntry;
import org.mozartspaces.core.MzsConstants;
import org.mozartspaces.core.RequestContext;

import sbc.group1.lab1.common.Order;

public class OrderCoordinator implements ImplicitNativeCoordinator {

    public final static String COORDINATOR_NAME = "OrderCoordinator";
    
    public static org.mozartspaces.capi3.Selector newSelector(final int count) {
        return new Selector(COORDINATOR_NAME, count);
    }
    
    public static org.mozartspaces.capi3.Selector newSelector(final int count, final int minAcceptedId) {
        return new Selector(COORDINATOR_NAME, count, minAcceptedId);
    }
    
    @Override
    public CoordinationData createDefaultCoordinationData() {
        return new CoordinationData() {
            
            @Override
            public String getName() {
                // TODO Auto-generated method stub
                return COORDINATOR_NAME;
            }
        };
    }

    private ConcurrentSkipListMap<Integer, NativeEntry> entries = 
        new ConcurrentSkipListMap<Integer, NativeEntry>();
    
    @Override
    public String getName() {
        return COORDINATOR_NAME;
    }

    @Override
    public void prepareEntryRemoval(NativeSubTransaction stx,
            NativeEntry entry, RequestContext context)
            throws CoordinatorLockedException {
    }

    @Override
    public boolean registerEntry(NativeSubTransaction stx,
            CoordinationData coordData, NativeEntry entry,
            RequestContext context) throws Capi3Exception {
        if (entry.getData() instanceof Order) {
            Order order = (Order) entry.getData();
            entries.put(order.getId(), entry);
            return true;
        }else {
            return false;
        }
    }

    @Override
    public void setIsolationManager(NativeIsolationManager isolationManager) {
    }

    @Override
    public boolean unregisterEntry(NativeEntry entry, RequestContext context) {
        if (entry.getData() instanceof Order) {
            Order order = (Order) entry.getData();
            return null != entries.remove(order.getId());
        }else {
            return false;
        }
    }
    
    ConcurrentSkipListMap<Integer, NativeEntry> getEntries() {
        return entries;
    }
    private static class Selector extends AbstractDefaultSelector {
        
        private OrderCoordinator coordinator;
        private ConcurrentSkipListMap<Integer, NativeEntry> entries;
        private Iterator<NativeEntry> iterator;
        private int currentCount;
        private int minAcceptedId;
        
        private Selector(final String name, final int count, final int minAcceptedId) {
            super(name, count);
            this.minAcceptedId=minAcceptedId;
        }
        
        private Selector(final String name, final int count) {
            this(name, count, Order.NO_ID);
        }

        @Override
        protected List<NativeEntry> getAllFirstSelector(final IsolationLevel isolationLevel,
                final NativeSubTransaction stx, final OperationType opType) throws CountNotMetException,
                EntryLockedException {

            checkCount(this.entries.size(), this.getCount(), this.getName());

            List<NativeEntry> result = new ArrayList<NativeEntry>();
            // iterate over all entries
            Iterator<NativeEntry> it;
            if (minAcceptedId==Order.NO_ID) {
                it = this.entries.values().iterator();
            }else {
                it = this.entries.tailMap(minAcceptedId).values().iterator();
            }
            
            while (it.hasNext()) {
                NativeEntry entry = it.next();
                if (isCountMet(result.size(), this.getCount())) {
                    return result;
                }
                // check whether the entry is available
                Availability availability = super.getIsolationManager().checkEntryAvailability(entry, isolationLevel,
                        stx, opType);
                switch (availability.getType()) {
                case AVAILABLE:
                    result.add(entry);
                    break;
                case NOTVISIBLE:
                    // insert lock on the entry
                    break;
                case NOTAVAILABLE:
                    // the operation would not be able to acquire a lock
                    if (this.getCount() == Selector.COUNT_MAX) {
                        // return all entries read until now
                        return result;
                    }
                    // Queue order conflict, COUNT_ALL not satisfiable
                    throw LockedExceptionsHelper.newEntryLockedException(availability);
                default:
                    throw new IllegalStateException("This state should not be reachable");
                }
            }
            checkCount(result.size(), this.getCount(), this.getName());
            return result;
        }

        @Override
        protected List<NativeEntry> getAllSelectorInside(final IsolationLevel isolationLevel,
                final NativeSubTransaction stx, final OperationType opType) throws CountNotMetException,
                EntryLockedException {

            // fetch all entries from previous stage
            List<NativeEntry> predecessorResult = super.getPredecessor().getAll(isolationLevel, stx, opType);
            checkCount(predecessorResult.size(), this.getCount(), this.getName());

            ConcurrentSkipListMap<Integer, NativeEntry> sorting = 
                new ConcurrentSkipListMap<Integer, NativeEntry>();
            for (NativeEntry entry : predecessorResult) {
                if (entry.getData() instanceof Order) {
                    Integer id = ((Order)entry.getData()).getId();
                    if (id > minAcceptedId) {
                        sorting.put(id, entry);
                    }
                }else {
                    continue;
                }
            }
            checkCount(sorting.size(), this.getCount(), this.getName());
            List<NativeEntry> result = new ArrayList<NativeEntry>(sorting.values());
            if (!MzsConstants.Selecting.isSpecialCountConstant(this.getCount())) {
                result = result.subList(0, this.getCount());
            }
            return result;
        }

        @Override
        protected NativeEntry getNextFirstSelector(final IsolationLevel isolationLevel, final NativeSubTransaction stx,
                final OperationType opType) throws CountNotMetException, EntryLockedException {

            if (isCountMet(currentCount, this.getCount())) {
                return null;
            }

            // ensure that we have an entry iterator
            if (this.iterator == null) {
                this.iterator = this.getAll(isolationLevel, stx, opType).iterator();
            }

            // get the next entry
            if (!this.iterator.hasNext()) {
                return null;
            }
            currentCount++;
            return this.iterator.next();
        }

        @Override
        protected NativeEntry getNextSelectorInside(final IsolationLevel isolationLevel,
                final NativeSubTransaction stx, final OperationType opType) throws CountNotMetException,
                EntryLockedException {
            return getNextFirstSelector(isolationLevel, stx, opType);
        }
        

        @Override
        public void link(NativeCoordinator coordinator,
                NativeIsolationManager isolationManager) {
            super.setIsolationManager(isolationManager);
            this.coordinator = (OrderCoordinator) coordinator;
            this.entries = this.coordinator.getEntries();
        }
        
    }
    

    public static final class CoordinatorTranslator implements
            NativeCoordinatorTranslator<OrderCoordinator, OrderCoordinator> {
        @Override
        public OrderCoordinator translateCoordinator(final OrderCoordinator coordinator) {
            return coordinator;
        }
    }


    public static final class SelectorTranslator implements
            NativeSelectorTranslator<OrderCoordinator.Selector, OrderCoordinator.Selector> {
        @Override
        public OrderCoordinator.Selector translateSelector(final OrderCoordinator.Selector selector) {
            return selector;
        }
    }

}
